#1390 - configurable storage for aggregating functions#2473
Open
MrHDOLEK wants to merge 10 commits into
Open
Conversation
GroupBy aggregation can move state out of the PHP heap instead of holding every result in memory. A new AggregationStorage abstraction offers three backends: memory (default, unchanged behavior), filesystem spill+merge, and PSR-16 key-value (APCu/Redis). - MergeableAggregatingFunction + merge() on all built-in aggregators - AggregationStorage: Memory / External (filesystem spill+merge) / Kv (PSR-16) - KV storage fails loudly on eviction / write failure instead of returning a wrong aggregate - config: AggregationStorageStrategy + FLOW_AGGREGATION_MAX_MEMORY wired into ConfigBuilder - APCu enabled in the nix dev shell (for the PSR-16/APCu storage test)
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## 1.x #2473 +/- ##
============================================
+ Coverage 85.97% 85.99% +0.02%
- Complexity 0 22337 +22337
============================================
Files 1662 1670 +8
Lines 68370 68583 +213
============================================
+ Hits 58783 58981 +198
- Misses 9587 9602 +15 🚀 New features to boost your workflow:
|
This comment has been minimized.
This comment has been minimized.
norberttech
reviewed
Jun 22, 2026
| */ | ||
| public function __construct( | ||
| public array $values, | ||
| public array $aggregators, |
Member
There was a problem hiding this comment.
I'm not really sure if I understand this data structure, it's keeping values for all aggregators?
norberttech
reviewed
Jun 22, 2026
| return $this; | ||
| } | ||
|
|
||
| public function aggregationStorage(AggregationStorageStrategy $strategy): self |
norberttech
reviewed
Jun 22, 2026
| return $this; | ||
| } | ||
|
|
||
| public function aggregationStore(AggregationStorage $storage): self |
Member
There was a problem hiding this comment.
Those two methods, aggregationStorage and aggregationStore are a bit confusing
norberttech
reviewed
Jun 22, 2026
|
|
||
| public function flush(): void; | ||
|
|
||
| public function save(string $hash, GroupEntry $entry): void; |
Member
There was a problem hiding this comment.
what's the difference between entry and save?
norberttech
reviewed
Jun 22, 2026
|
|
||
| namespace Flow\ETL\Function; | ||
|
|
||
| interface MergeableAggregatingFunction extends AggregatingFunction |
Member
There was a problem hiding this comment.
could you walk me through this logic? Like when two AggregatingFunctions can be merged?
…torage Reworks the aggregation-storage approach following review. External aggregation is now sort-then-fold: sort rows by the group-by columns (reusing ExternalSort, which spills and merges in bounded memory) and fold adjacent equal keys. This makes the whole abstraction simpler and drops the concepts the review found confusing. - remove the AggregationStorage layer (Memory/External/Kv storage, GroupEntry, GroupBucket(sCache)) and the AggregationStorageStrategy enum - remove MergeableAggregatingFunction and merge() from aggregators - sort-fold never splits a group, so no merge contract is needed - add AggregatingAlgorithm with MemoryAggregation (default, hash map) and ExternalAggregation (sort + fold), selected via AggregationAlgorithms (mirrors SortAlgorithms) - GroupBy is now a spec (references/aggregations/hash/keyValues); GroupByProcessor routes pivot and global aggregation to memory - config: aggregationAlgorithm() replaces aggregationStorage()/aggregationStore(); output is streamed in batches instead of one Rows - external tags rows with an input ordinal and sorts by (group keys, ordinal), so first()/last()/collect() return the same result as the in-memory path even when a group spills across sort runs
Non-pivot aggregation now runs through hash aggregation with adaptive spilling: groups accumulate in an in-memory hash map, and once the configured memory limit is exceeded the map is spilled to a hash-sorted bucket and cleared. If nothing spills the map is the result; otherwise a streaming k-way merge folds partial aggregates by group key, so spilled state is O(groups), not O(rows). - merge() moves onto the base AggregatingFunction contract (no separate opt-in interface); every built-in aggregator implements it and self-validates by reference and result-affecting arguments - HashAggregation + GroupState / BucketGroup / GroupsMinHeap / GroupBucketsCache (FilesystemGroupBucketsCache), mirroring the external-sort building blocks - GroupBy becomes a spec (references/aggregations/hash/keyValues/newAggregators); pivot stays memory-only - config: aggregationMemoryLimit(Unit) (default unbounded, opt-in spilling) replaces the algorithm enum; output streamed in batches - fix multi-column group-by key hash collision (length-prefix the values)
- guard strpos()===false and drop unused mago pragmas flagged by CI static analysis
- annotate the bucket-cursor generators as Generator<int, array{...}> and narrow current() in the k-way merge
- cover merge() on all 10 aggregators (success, null branches, self-validation), the min-heap ordering/guard, the bucket cache round-trip, and the memory-limit env var
- drop the unused mago pragma in Sum::merge - hoist the pivot prototype behind an instanceof guard so clone is never applied to a possibly-false current() - drop over-specific @return docblocks that toArray() cannot satisfy
norberttech
reviewed
Jul 7, 2026
| private function hotCursor(array $hot): Generator | ||
| { | ||
| foreach ($hot as $hash => $state) { | ||
| yield [$hash, $state]; |
Member
There was a problem hiding this comment.
please don't use tuples for return types, introduce a simple VO instead
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Makes GroupBy aggregation memory-efficient and configurable (issue #1390): keep the fast in-memory path by default, or opt into an external, bounded-memory path.
The external path is sort-then-fold: sort the rows by the group-by columns (reusing
ExternalSort, which already spills to disk and merges in bounded memory), then fold adjacent equal keys in a single pass. Because a group is never split across independent partial accumulators, there is no merge contract — the whole thing is just "sort, then collapse equal keys", mirroring howExternalSortworks. Output is streamed in batches instead of one bigRows.aggregationAlgorithm(AggregationAlgorithms::MEMORY_AGGREGATION | EXTERNAL_AGGREGATION)selects the strategy (mirrorsSortConfigBuilder::algorithm()); the default stays in-memory, so existing pipelines are unchanged.Addressing the review
GroupEntry"keeps values for all aggregators?" — gone. There is no storage layer; the algorithms hold their accumulators directly.aggregationStorage(strategy)— "storage or strategy?" andaggregationStorage()vsaggregationStore()— replaced by a singleaggregationAlgorithm(AggregationAlgorithms); the "storage/store" naming is removed.AggregationStorage::entry()vssave()— gone with the storage interface; aggregation is one inline loop.MergeableAggregatingFunction"when can two functions merge?" — removed entirely. Sort-fold never merges partial aggregates, so the interface andmerge()on the built-ins are deleted.Correctness note: the external path tags each row with its input position and sorts by
(group keys, input ordinal), sofirst()/last()/ unsortedcollect()return the same result as the in-memory path even when a group spills across sort runs (covered by a test).Resolves: #1390
Change Log
Added
ConfigBuilder::aggregationAlgorithm()and theAggregationAlgorithmsenum select betweenMEMORY_AGGREGATION(default, in-memory hash map) andEXTERNAL_AGGREGATION(sorts by the group-by columns and folds, spilling to disk for bounded memory), withaggregationFilesystem()for the spill protocol.Rows.Fixed
Changed
Removed
Deprecated
Security